/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.analysis.restapi.controllers;

import eu.dicodeproject.analysis.restapi.SchemaConfig;
import twitter4j.internal.org.json.JSONArray;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;

/**
 * @author jakob
 */
public class VectorsController extends FilteringController {

  private Map<String,SchemaConfig> schemaConfigMap;

  /**
   * Returns term vectors for documents in a time range and other filters.
   * @param source source table (e.g. twitter or blogs)
   * @param start start date (inclusive) for time range
   * @param stop stop date (exclusive) for time range
   * @param language language to filter by
   * @param regex regular expression that has to match parts of the document text
   * @param limit user-specific result limit; cannot be larger than maximumResultLimit
   * @param callback JSON callback string
   * @param response object to write the result to
   * @throws java.io.IOException inherited from HBase Scan and HttpServletResponse.write
   * @throws twitter4j.internal.org.json.JSONException inherited from JSONObject constructor
   */
  @RequestMapping(value = "{source}/vectors", method = RequestMethod.GET)
  public void vectors(@PathVariable String source,
      @RequestParam(required = false) String start, @RequestParam(required = false) String stop,
      @RequestParam(required = false) String language, @RequestParam(required = false) String regex,
      @RequestParam(required = false) Integer limit, @RequestParam(required = false) String callback,
      HttpServletResponse response) throws IOException, JSONException {

    // get schema
    SchemaConfig schemaConfig = schemaConfigMap.get(source);

    // collect parameters for result
    JSONObject params = new JSONObject();

    // set HBase Scan
    Scan scan = getScan(schemaConfig, start, stop, language, regex, params);

    // get HBase table
    HTableInterface hTable = getHTable(schemaConfig.getTableName());

    // get result limit
    int resultLimit = getResultLimit(limit);
    params.put("limit", resultLimit);

    // read results from scan into a list of JSONObjects
    int totalResults = 0;

    List<JSONObject> vectorsList = new ArrayList<JSONObject>();
    
    for (Result result : hTable.getScanner(scan)) {
    
      JSONObject doc=null;
      String id = Bytes.toString(result.getRow());
      id = id.substring(schemaConfig.idStartPositionInKey());  // skip creation date

      KeyValue nerKv = result.getColumnLatest(schemaConfig.fhgColumnFamily(), schemaConfig.nerColumn());
      
      // Entities are stored in a JSON formatted map in HBase
      // FIXME: change format in HBase to get rid of ugly renaming
      // FIXME: check for empty NEs and vectors and remove quotation marks and newlines before storing NEs/vectors
      if (nerKv != null) {
        try {
          doc = new JSONObject(Bytes.toString(nerKv.getValue()));
          JSONArray nes = (JSONArray) doc.get("result");
          
          if (nes.length() == 0) {
            doc = new JSONObject(); // reset
          }
          else {
            doc = new JSONObject();
            doc.put("n", nes);
          }
        } catch (JSONException e) {
          log.info("could not parse JSON string: "+Bytes.toString(nerKv.getValue())+e.getMessage()+"\n");
        } 
      } 
      else {
        doc = new JSONObject();
        log.debug(String.format("named entities not processed yet in row %s column %s:%s ", id, Bytes.toString(schemaConfig.fhgColumnFamily()), Bytes.toString(schemaConfig.nerColumn())));        
      }
      
      // there might be no object yet because of exception
      if (doc==null)  { 
        doc = new JSONObject();
      }
      
      doc.put("id", id);
      KeyValue vectorKv = result.getColumnLatest(schemaConfig.fhgColumnFamily(), schemaConfig.contentColumn());
      
      if (vectorKv == null) {
        log.debug(String.format("did not find vector in row %s column %s:%s ", id, Bytes.toString(schemaConfig.columnFamily()), Bytes.toString(schemaConfig.contentColumn())));        
      }      
      else {
        doc.put("v", Bytes.toString(vectorKv.getValue()));
      }

      // empty entry? 
      if (nerKv == null && vectorKv == null) continue;
      
      vectorsList.add(doc);

      totalResults++;
      if (totalResults >= resultLimit) {
        break;
      }
    }
    params.put("results", totalResults);

    // put result list and parameters map into a JSON
    JSONObject json = new JSONObject();
    json.put("parameters", params);
    json.put("vectors", vectorsList);

    // write out the result JSON
    respond(callback, json, response);
    log.info("Returning " + totalResults + " for query: "+ regex + " language: "+language+ " start: "+start+ " ,end: "+stop);
  }

  public void setSchemaConfigMap(Map<String, SchemaConfig> schemaConfigMap) {
    this.schemaConfigMap = schemaConfigMap;
  }

}